package com.haishuibutian.rocketmqconsumer.rocketmq;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;


@Component
public class MQListener implements MessageListenerConcurrently {

    private static final Logger logger = LoggerFactory.getLogger(MQListener.class);

    private final MQConfig mqConfig;

    private final Map<String, MQConsumerHandler<Object>> handlerMap = new ConcurrentHashMap<>();

    public MQListener(MQConfig mqConfig) {
        this.mqConfig = mqConfig;
    }


    public void registerHandler(String tags, MQConsumerHandler<Object> handler) {
        handlerMap.put(tags, handler);
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        logger.info("Consumer receive message size = {}", msgs.size());
        String json = null;
        try {

            for (MessageExt msg : msgs) {
                json = new String(msg.getBody());
                logger.info("Consumer message: {}, topic: {}, tag: {}, key: {}, reconsumeTimes: {}", json, msg.getTopic(), msg.getTags(), msg.getKeys(),
                        msg.getReconsumeTimes());
                final String tags = msg.getTags();
                final MQConsumerHandler handler = handlerMap.get(tags);
                if (Objects.isNull(handler)){
                    logger.warn("tag: {} 未找到对应的消费类 无效消费",tags);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                final Object transferMessage = handler.transferMessage(json);
                boolean result = handler.handler(transferMessage);
                if (!result) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }

        } catch (Exception e) {
            logger.error(json, e);
            if (msgs.get(0).getReconsumeTimes() == mqConfig.getMaxReConsumeTimes()) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
